package com.west.xu.tdmq.config;

import com.west.xu.tdmq.handle.ConsumerDeadLetterMsg;
import com.west.xu.tdmq.handle.ConsumerMsg;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.pulsar.client.api.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

@Configuration
@ConfigurationProperties(prefix = "tdmq")
public class PulsarConfig {

    private static final Log logger = LogFactory.getLog(PulsarConfig.class);

    @Resource
    private PulsarProperties pulsarProperties;
    /**
     * 订阅模式：
     * 1、独享（exclusive）订阅——同时只能有一个消费者。
     * 2、共享（shared）订阅——可以由多个消费者订阅，每个消费者接收其中的一部分消息。
     * 3、失效备援（failover）订阅——允许多个消费者连接到同一个主题上，但只有一个消费者能够接收消息。
     * 只有在当前消费者发生失效时，其他消费者才开始接收消息
     *
     * @return
     * @throws PulsarClientException
     */
    /**
     *
     */
    @Value("${tdmq.notice.topic}")
    private String noticeTopic;
    @Value("${tdmq.notice.consumerName}")
    private String noticeConsumerName;
    @Value("${tdmq.notice.subscriptionName}")
    private String noticeSubscriptionName;
    @Value("${tdmq.notice.maxRedeliveryCount}")
    private Integer noticeMaxRedeliveryCount;
    @Value("${tdmq.notice.ackTimeOut}")
    private Long noticeAckTimeOut;
    /**
     *
     */
    @Value("${tdmq.notice.deadLetter.topic}")
    private String noticeDeadLetterTopic;
    @Value("${tdmq.notice.deadLetter.consumerName}")
    private String noticeDeadLetterConsumerName;
    @Value("${tdmq.notice.deadLetter.subscriptionName}")
    private String noticeDeadLetterSubscriptionName;

    /**
     * @return
     * @throws PulsarClientException
     */
    @Bean(name = "pulsarClient")
    public PulsarClient pulsarClient() throws PulsarClientException {
        return PulsarClient.builder()
                .serviceUrl(pulsarProperties.getServiceUrl())
//                .listenerName(pulsarProperties.getListenerName())
//                .authentication(AuthenticationFactory.token(pulsarProperties.getToken()))
                .build();
    }

    /**
     * @return
     * @throws PulsarClientException
     */
    @Bean(name = "noticeConsumer")
    public Consumer noticeConsumer() throws PulsarClientException {
        return pulsarClient()
                .newConsumer()
                .consumerName(noticeConsumerName.concat("-1"))
                .topic(noticeTopic)
                .subscriptionName(noticeSubscriptionName)
                .messageListener(new ConsumerMsg<>())
                .ackTimeout(noticeAckTimeOut, TimeUnit.SECONDS)               //确认超时时间
                .subscriptionType(SubscriptionType.Shared)       //设置订阅模式
                .deadLetterPolicy(DeadLetterPolicy.builder()
                        //.deadLetterTopic("Notice-west-dlq")
                        .maxRedeliverCount(2)
                        .build())
//                .enableRetry(true)      //是否可重试
                .batchReceivePolicy(BatchReceivePolicy.builder().maxNumMessages(17).build())
                .subscribe();
    }

    /**
     * @return
     * @throws PulsarClientException
     */
    @Bean(name = "noticeDeadLetterConsumer")
    public Consumer noticeDeadLetterConsumer() throws PulsarClientException {
        return pulsarClient()
                .newConsumer()
//                .topic("Notice-west-dlq")
                .topic(noticeDeadLetterTopic)
                .consumerName(noticeDeadLetterConsumerName)
                .subscriptionName(noticeDeadLetterSubscriptionName)
                .messageListener(new ConsumerDeadLetterMsg<>())
                .ackTimeout(noticeAckTimeOut, TimeUnit.SECONDS)               //确认超时时间
                .subscriptionType(SubscriptionType.Shared)       //设置订阅模式
                .subscribe();
    }
}
